package com.example.hlsdemo.service.impl;

import com.example.hlsdemo.entity.VideoInfo;
import com.example.hlsdemo.mapper.VideoInfoMapper;
import com.example.hlsdemo.service.VideoService;
import com.example.hlsdemo.utils.FFmpegUtil;
import io.minio.*;
import io.minio.messages.DeleteError;
import io.minio.messages.DeleteObject;
import io.minio.messages.Item;
import lombok.extern.slf4j.Slf4j;
import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;

import java.io.*;
import java.nio.file.*;
import java.nio.file.attribute.BasicFileAttributes;
import java.security.MessageDigest;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

@Slf4j
@Service
public class VideoServiceImpl implements VideoService {

    private static final String TMP_DIR = "video-pro-";

    private static final String BUCKET_NAME = "videos";
    private final ExecutorService executorService;

    @Autowired
    private MinioClient minioClient;

    @Autowired
    private VideoInfoMapper videoInfoMapper;

    /**
     * 构造函数
     * 初始化线程池，用于并行处理视频上传和转换
     * 核心线程数 = CPU核心数
     * 最大线程数 = CPU核心数 * 2
     * 使用守护线程，确保主程序退出时线程池也会退出
     */
    public VideoServiceImpl() {
        int processors = Runtime.getRuntime().availableProcessors();
        this.executorService = new ThreadPoolExecutor(
                processors,
                processors * 2,
                60L,
                TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(1000),
                new ThreadFactory() {
                    private final AtomicInteger threadNumber = new AtomicInteger(1);

                    @Override
                    public Thread newThread(Runnable r) {
                        Thread t = new Thread(r, TMP_DIR + threadNumber.getAndIncrement());
                        t.setDaemon(true);
                        return t;
                    }
                },
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }

    /**
     * 上传视频并转换为HLS格式
     *
     * @param file
     * @return videoId
     * @throws Exception
     */
    @Transactional
    @Override
    public String uploadAndConvertVideo(MultipartFile file) throws Exception {
        // 查询系统的bucket是否存在
        boolean bucketExists = minioClient.bucketExists(BucketExistsArgs.builder().bucket(BUCKET_NAME).build());
        if (!bucketExists) {
            // 如果不存在，则创建一个新的bucket
            minioClient.makeBucket(MakeBucketArgs.builder().bucket(BUCKET_NAME).build());
        }

        // 计算文件的MD5值防止重复提交文件
        String fileMd5 = calculateMD5(file.getInputStream());

        // 查询数据库中的md5值是否存在
        VideoInfo exits = videoInfoMapper.findByMd5(fileMd5);
        if (exits != null) {
            log.info("文件已经存在");
            return exits.getVideoId();
        }
        // 文件不存在
        String originalFilename = file.getOriginalFilename();
        String extensionFile = originalFilename.substring(originalFilename.lastIndexOf("."));
        String uniqueId = UUID.randomUUID().toString();
        String outputName = uniqueId + extensionFile;

        // 将上传的文件保存到临时目录中
        Path tempDirectory = Files.createTempDirectory(TMP_DIR);
        File tempFile = tempDirectory.resolve(outputName).toFile();
        //copy
        file.transferTo(tempFile);


        // 使用FFmpegUtils对文件进行不同清晰度切片处理
        boolean isSuccess = FFmpegUtil.convertToHLS(tempFile, tempDirectory, uniqueId, executorService);
        if (!isSuccess) {
            return null;
        }


        // 将临时文件夹中的hls上传到minio中
        CountDownLatch uploadHls = new CountDownLatch(1);
        // 记录总文件数和上传成功数量
        AtomicInteger uploadCount = new AtomicInteger(0);
        AtomicInteger successCount = new AtomicInteger(0);

        Path hlsDir = tempDirectory.resolve(FFmpegUtil.HLS_DIR);
        Files.walk(hlsDir).filter(Files::isRegularFile).forEach(path -> uploadCount.incrementAndGet());

        // 并行上传所有的ts文件和m3u8文件
        try (Stream<Path> paths = Files.walk(hlsDir)) {
            paths.filter(Files::isRegularFile)
                    .forEach(path -> {
                        executorService.submit(() -> {
                            try (FileInputStream fis = new FileInputStream(path.toFile())) {
                                String objName = uniqueId + "/" + path.getFileName().toString();
                                minioClient.putObject(
                                        PutObjectArgs.builder()
                                                .bucket(BUCKET_NAME)
                                                .object(objName)
                                                .stream(fis, Files.size(path), -1)
                                                .build()
                                );
                                int success = successCount.incrementAndGet();
                                if (success == uploadCount.get()) {
                                    uploadHls.countDown();
                                }
                            } catch (Exception e) {
                                log.error("上传文件到minio中失败");
                                uploadHls.countDown();
                            }
                        });
                    });
        }

        // 等待所有文件上传成功（30分钟）
        boolean await = uploadHls.await(30, TimeUnit.MINUTES);
        if (!await) {
            log.error("文件上传超时");
        }
        //将文件数据保存到数据库中
        VideoInfo videoInfo = new VideoInfo();
        videoInfo.setVideoId(uniqueId);
        videoInfo.setFileName(originalFilename);
        videoInfo.setMd5(fileMd5);
        videoInfo.setBucketName(BUCKET_NAME);
        videoInfo.setM3u8Path(uniqueId + "/" + uniqueId + "_mid.m3u8"); // 默认使用中等版本
        videoInfo.setOriginalM3u8Path(uniqueId + "/" + uniqueId + "_original.m3u8");
        videoInfo.setLowM3u8Path(uniqueId + "/" + uniqueId + "_low.m3u8");
        videoInfo.setMidM3u8Path(uniqueId + "/" + uniqueId + "_mid.m3u8");
        videoInfo.setHighM3u8Path(uniqueId + "/" + uniqueId + "_high.m3u8");
        videoInfo.setStatus("COMPLETED"); // 同步处理完成，设置为完成状态
        videoInfo.setFileSize(file.getSize());
        videoInfo.setContentType(file.getContentType());
        videoInfo.setCreateTime(LocalDateTime.now());
        videoInfo.setUpdateTime(LocalDateTime.now());
        videoInfoMapper.insert(videoInfo);
        // 清理临时文件
        cleanupTempFiles(tempDirectory);
        return uniqueId;
    }

    /**
     * 清理临时文件
     * 使用重试机制确保文件被删除
     */
    private void cleanupTempFiles(Path tempDir) {
        try {
            TimeUnit.SECONDS.sleep(1);
            deleteDirectory(tempDir);
        } catch (Exception e) {
            log.error("清理临时目录失败: " + tempDir, e);
        }
    }


    /**
     * 获取视频播放列表
     * 客户端首先获取.m3u8文件，然后根据播放列表按需请求.ts分片
     */
    @Override
    public byte[] getPlaylist(String videoId, String playlistName) throws Exception {
        try (InputStream is = minioClient.getObject(
                GetObjectArgs.builder()
                        .bucket(BUCKET_NAME)
                        .object(playlistName)
                        .build())) {
            return is.readAllBytes();
        }
    }

    /**
     * 获取视频分片文件
     * 客户端根据播放列表中的信息请求具体的.ts分片
     */
    @Override
    public byte[] getVideoFile(String videoId, String filename) throws Exception {
        try (InputStream is = minioClient.getObject(
                GetObjectArgs.builder()
                        .bucket(BUCKET_NAME)
                        .object(videoId + "/" + filename)
                        .build())) {
            return is.readAllBytes();
        }
    }

    /**
     * 删除目录及其内容
     * 使用重试机制确保文件被删除
     */
    private void deleteDirectory(Path directory) throws IOException {
        int maxRetries = 3;
        int retryCount = 0;
        while (retryCount < maxRetries) {
            try {
                Files.walkFileTree(directory, new SimpleFileVisitor<>() {
                    @NotNull
                    @Override
                    public FileVisitResult visitFile(Path file, @NotNull BasicFileAttributes attrs) throws IOException {
                        try {
                            Files.delete(file);
                        } catch (IOException e) {
                            log.warn("删除文件失败: {}, 重试中...", file);
                            try {
                                TimeUnit.MILLISECONDS.sleep(100);
                                Files.delete(file);
                            } catch (Exception ex) {
                                log.error("重试删除文件失败: {}", file, ex);
                            }
                        }
                        return FileVisitResult.CONTINUE;
                    }

                    @NotNull
                    @Override
                    public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException {
                        try {
                            Files.delete(dir);
                        } catch (IOException e) {
                            log.warn("删除目录失败: {}, 重试中...", dir);
                            try {
                                TimeUnit.MILLISECONDS.sleep(100);
                                Files.delete(dir);
                            } catch (Exception ex) {
                                log.error("重试删除目录失败: {}", dir, ex);
                            }
                        }
                        return FileVisitResult.CONTINUE;
                    }
                });
                break;
            } catch (IOException e) {
                retryCount++;
                if (retryCount == maxRetries) {
                    throw e;
                }
                try {
                    TimeUnit.SECONDS.sleep(1);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }

    /**
     * 计算文件的MD5值
     * 用于文件去重
     */
    private String calculateMD5(InputStream inputStream) throws Exception {
        MessageDigest md = MessageDigest.getInstance("MD5");
        byte[] buffer = new byte[8192];
        int read;
        while ((read = inputStream.read(buffer)) > 0) {
            md.update(buffer, 0, read);
        }
        byte[] digest = md.digest();
        StringBuilder sb = new StringBuilder();
        for (byte b : digest) {
            sb.append(String.format("%02x", b));
        }
        return sb.toString();
    }

    @Override
    public List<VideoInfo> getAllVideos() {
        return videoInfoMapper.selectList(null);
    }

    /**
     * 异步上传和转码视频
     * 1. 快速处理：直接将原始视频转为HLS格式（不转码，只切片）速度快
     * 2. 异步处理：在后台进行三种清晰度的转码
     * 3. 用户可以立即观看原始视频，在转码完成后切换到对应清晰度
     *
     * @param file 上传的视频文件
     * @return 视频ID
     */
    @Transactional
    @Override
    public String uploadVideoAsync(MultipartFile file) throws Exception {
        // 确保bucket存在
        boolean found = minioClient.bucketExists(BucketExistsArgs.builder().bucket(BUCKET_NAME).build());
        if (!found) {
            minioClient.makeBucket(MakeBucketArgs.builder().bucket(BUCKET_NAME).build());
        }

        // 计算文件的MD5值用于去重
        String fileMd5 = calculateMD5(file.getInputStream());

        // 检查文件是否已存在
        VideoInfo existingVideo = videoInfoMapper.findByMd5(fileMd5);
        if (existingVideo != null) {
            log.info("文件已存在，直接返回视频ID: {}", existingVideo.getVideoId());
            return existingVideo.getVideoId();
        }

        // 生成唯一文件名
        String originalFilename = file.getOriginalFilename();
        if (originalFilename == null || originalFilename.isEmpty()) {
            throw new IllegalArgumentException("文件名不能为空");
        }

        String extension = originalFilename.substring(originalFilename.lastIndexOf("."));
        String uniqueId = UUID.randomUUID().toString();
        String outputName = uniqueId + extension;

        // 创建临时目录保存上传的文件
        Path tempDir = Files.createTempDirectory(TMP_DIR + uniqueId);
        File tempFile = tempDir.resolve(outputName).toFile();

        try {
            // 保存文件到临时目录
            file.transferTo(tempFile);

            // 1. 快速处理：立即将原始视频转为HLS格式（不转码，只切片）
            boolean success = FFmpegUtil.convertToOriginalHLS(tempFile, tempDir, uniqueId, executorService);
            if (!success) {
                log.error("快速处理视频失败");
            }

            // 上传原始HLS文件到MinIO
            Path hlsDir = tempDir.resolve(FFmpegUtil.HLS_DIR);
            uploadHLSFilesWithPattern(hlsDir, uniqueId, "_original");

            // 保存视频信息到数据库
            VideoInfo videoInfo = new VideoInfo();
            videoInfo.setVideoId(uniqueId);
            videoInfo.setFileName(originalFilename);
            videoInfo.setMd5(fileMd5);
            videoInfo.setBucketName(BUCKET_NAME);
            videoInfo.setM3u8Path(uniqueId + "/" + uniqueId + "_mid.m3u8"); // 默认使用中版本
            videoInfo.setOriginalM3u8Path(uniqueId + "/" + uniqueId + "_original.m3u8");
            videoInfo.setLowM3u8Path(uniqueId + "/" + uniqueId + "_low.m3u8");
            videoInfo.setMidM3u8Path(uniqueId + "/" + uniqueId + "_mid.m3u8");
            videoInfo.setHighM3u8Path(uniqueId + "/" + uniqueId + "_high.m3u8");
            videoInfo.setStatus("PROCESSING"); // 同步处理完成后直接设置为完成状态
            videoInfo.setFileSize(file.getSize());
            videoInfo.setContentType(file.getContentType());
            videoInfo.setCreateTime(LocalDateTime.now());
            videoInfo.setUpdateTime(LocalDateTime.now());
            videoInfoMapper.insert(videoInfo);

            // 2. 异步处理：在后台进行三种清晰度的转码
            final Path finalTempDir = tempDir;
            FFmpegUtil.asyncConvertMultipleResolutions(tempFile, tempDir, uniqueId, executorService, () -> {
                try {
                    // 转码完成后的处理
                    log.info("视频ID: {} 的多清晰度转码完成，开始上传", uniqueId);

                    // 上传转码后的HLS文件到MinIO
                    uploadHLSFilesWithPattern(finalTempDir.resolve("hls"), uniqueId, "_high|_mid|_low");

                    // 更新数据库记录
                    VideoInfo updatedInfo = videoInfoMapper.findById(uniqueId);
                    if (updatedInfo != null) {
                        updatedInfo.setM3u8Path(uniqueId + "/" + uniqueId + "_high.m3u8"); // 更新为高清m3u8
                        updatedInfo.setLowM3u8Path(uniqueId + "/" + uniqueId + "_low.m3u8");
                        updatedInfo.setMidM3u8Path(uniqueId + "/" + uniqueId + "_mid.m3u8");
                        updatedInfo.setHighM3u8Path(uniqueId + "/" + uniqueId + "_high.m3u8");
                        updatedInfo.setStatus("COMPLETED"); // 标记为完成
                        updatedInfo.setUpdateTime(LocalDateTime.now());
                        videoInfoMapper.updateById(updatedInfo);
                    }

//                    removeOriginalHLSFiles(uniqueId);

                    // 清理临时文件
                    cleanupTempFiles(finalTempDir);

                    log.info("视频ID: {} 的异步处理全部完成", uniqueId);
                } catch (Exception e) {
                    log.error("转码后处理失败: " + uniqueId, e);
                }
            });
            return uniqueId;

        } catch (Exception e) {
            log.error("处理视频文件失败", e);
            cleanupTempFiles(tempDir);
            throw e;
        }
    }

    /**
     * 上传指定模式的HLS文件到MinIO
     */
    private void uploadHLSFilesWithPattern(Path hlsDir, String uniqueId, String pattern) throws Exception {
        CountDownLatch uploadLatch = new CountDownLatch(1);
        AtomicInteger successCount = new AtomicInteger(0);
        AtomicInteger totalFiles = new AtomicInteger(0);

        // 计算需要上传的总文件数
        Files.walk(hlsDir)
                .filter(Files::isRegularFile)
                .filter(path -> {
                    String filename = path.getFileName().toString();
                    return filename.matches(".*(" + pattern + ").*\\.(?:m3u8|ts)$");
                })
                .forEach(path -> totalFiles.incrementAndGet());

        if (totalFiles.get() == 0) {
            log.warn("没有找到匹配模式 {} 的HLS文件", pattern);
            return;
        }

        // 并行上传所有匹配的文件
        Files.walk(hlsDir)
                .filter(Files::isRegularFile)
                .filter(path -> {
                    String filename = path.getFileName().toString();
                    return filename.matches(".*(" + pattern + ").*\\.(?:m3u8|ts)$");
                })
                .forEach(path -> executorService.submit(() -> {
                    try (FileInputStream fis = new FileInputStream(path.toFile())) {
                        String objectName = uniqueId + "/" + path.getFileName().toString();
                        minioClient.putObject(
                                PutObjectArgs.builder()
                                        .bucket(BUCKET_NAME)
                                        .object(objectName)
                                        .stream(fis, Files.size(path), -1)
                                        .build()
                        );
                        // 更新上传进度
                        int currentSuccess = successCount.incrementAndGet();

                        if (currentSuccess == totalFiles.get()) {
                            uploadLatch.countDown();
                        }
                    } catch (Exception e) {
                        log.error("上传文件失败: " + path, e);
                        uploadLatch.countDown(); // 发生错误时释放等待
                    }
                }));

        // 等待所有文件上传完成（30分钟）
        boolean completed = uploadLatch.await(30, TimeUnit.MINUTES);
        if (!completed) {
            throw new RuntimeException("文件上传超时");
        }

        log.info("所有匹配模式 {} 的HLS文件上传完成", pattern);
    }

    /**
     * 从MinIO中删除原始HLS文件
     */
    private void removeOriginalHLSFiles(String videoId) {
        try {
            // 列出匹配模式的所有对象
            String prefix = videoId + "/" + videoId + "_original";

            Iterable<Result<Item>> results = minioClient.listObjects(
                    ListObjectsArgs.builder()
                            .bucket(BUCKET_NAME)
                            .prefix(prefix)
                            .recursive(true)
                            .build());

            // 收集所有要删除的对象
            List<DeleteObject> objects = new ArrayList<>();
            for (Result<Item> result : results) {
                Item item = result.get();
                objects.add(new DeleteObject(item.objectName()));
            }

            // 批量删除
            if (!objects.isEmpty()) {
                Iterable<Result<DeleteError>> results2 = minioClient.removeObjects(
                        RemoveObjectsArgs.builder()
                                .bucket(BUCKET_NAME)
                                .objects(objects)
                                .build());

                // 检查是否有删除错误
                for (Result<DeleteError> result : results2) {
                    DeleteError error = result.get();
                    log.error("删除原始文件失败: {}, 错误: {}", error.objectName(), error.message());
                }
            }
        } catch (Exception e) {
            log.error("删除原始HLS文件出错", e);
        }
    }

    /**
     * 获取视频信息
     *
     * @param videoId 视频ID
     * @return 视频信息
     */
    public VideoInfo getVideoInfo(String videoId) {
        return videoInfoMapper.findById(videoId);
    }
}